package org.budo.dubbo.protocol.async.repository.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.MQPullConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.budo.dubbo.protocol.async.repository.AbstractAsyncRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;

import lombok.Getter;
import lombok.Setter;

/**
 * @author limingwei
 */
@Getter
@Setter
public abstract class AbstractRocketMQAsyncRepository extends AbstractAsyncRepository implements DisposableBean {
    private static final Logger log = LoggerFactory.getLogger(AbstractRocketMQAsyncRepository.class);

    private String nameServerAddress; // "localhost:9876"

    private String group; // "please_rename_unique_group_name"

    private MQProducer rocketMQProducer;

    private MQPullConsumer rocketMQPullConsumer;

    protected MQProducer getRocketMQProducer() {
        if (null != this.rocketMQProducer) {
            return this.rocketMQProducer;
        }

        synchronized (RocketMQAsyncRepository.class) {
            if (null != this.rocketMQProducer) {
                return this.rocketMQProducer;
            }

            return this.rocketMQProducer = this.initRocketMQProducer();
        }
    }

    protected MQPullConsumer getRocketMQPullConsumer() {
        if (null != this.rocketMQPullConsumer) {
            return this.rocketMQPullConsumer;
        }

        synchronized (RocketMQAsyncRepository.class) {
            if (null != this.rocketMQPullConsumer) {
                return this.rocketMQPullConsumer;
            }

            return this.rocketMQPullConsumer = this.initRocketMQPullConsumer();
        }
    }

    private MQPullConsumer initRocketMQPullConsumer() {
        try {
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(this.getGroup());
            consumer.setNamesrvAddr(this.getNameServerAddress());

            consumer.start();

            return consumer;
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    private MQProducer initRocketMQProducer() {
        try {
            DefaultMQProducer producer = new DefaultMQProducer(this.getGroup());
            producer.setNamesrvAddr(this.getNameServerAddress());

            producer.start();

            return producer;
        } catch (Throwable e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void destroy() throws Exception {
        try {
            MQProducer rocketMQProducer = this.getRocketMQProducer();
            rocketMQProducer.shutdown();
        } catch (Throwable e) {
            log.error("#86 rocketMQProducer.shutdown error, e=" + e, e);
        }

        try {
            MQPullConsumer rocketMQPullConsumer = this.getRocketMQPullConsumer();
            rocketMQPullConsumer.shutdown();
        } catch (Throwable e) {
            log.error("#93 rocketMQProducer.shutdown error, e=" + e, e);
        }
    }
}